package easyProductConsumerDemo;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @program: KafkaSparkStreaming
 * @description
 * @author: tkk fendoukaoziji
 * @create: 2019-05-24 16:13
 **/
public class ConsumerDemo {
    public static final String brokerList="node2:9092,node3:9092,node4:9092";
    public static final String topic="topic-kafkaBook";
    public static final String groupId="group.tukk";
    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("bootstrap.servers",brokerList);
        properties.put("group.id",groupId);
        //创建消费者客户端实例
        KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);
        //订阅主题
        consumer.subscribe(Collections.singleton(topic));
        //循环消费消息
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String,String> record :records ) {
                System.out.println(record);
                System.out.println(record.value());
            }
        }

    }
}










